package com.atguigu.api

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * @ClassName SideOutputTest
 * @Description
 * @Author Mr Yang
 * @Date 2020/9/8 21:17
 * @Version 1.0
 */
object SideOutputTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.socketTextStream("localhost", 7777)
    val dataStream = stream.map( data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
      override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
    })

    val processStream = dataStream.process(new FreezingAlert())

    processStream.getSideOutput(new OutputTag[String]("freezing alert")).print("alert data")
    processStream.print("process data")
    env.execute("sideOutput test")
  }
}

//冰点报警 如果小于32F，输出报警信息到侧输出流
// I-输入流, O-主输出流数据
class FreezingAlert() extends ProcessFunction[SensorReading,SensorReading]{

  lazy val outputTag: OutputTag[String] = new OutputTag[String]("freezing alert")

  override def processElement(i: SensorReading, context: ProcessFunction[SensorReading, SensorReading]#Context, collector: Collector[SensorReading]): Unit = {
    if(i.temperature < 32.0){
      //侧输出流
      context.output(outputTag, "freezing alert for " + i.id)
    }else{
      //主输出流
      collector.collect(i)
    }
  }
}